package kalp.web;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

import javax.annotation.PostConstruct;

import org.apache.spark.api.java.function.Function;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.twitter.TwitterUtils;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import twitter4j.Status;
import twitter4j.auth.Authorization;
import twitter4j.auth.OAuthAuthorization;
import twitter4j.conf.Configuration;
import twitter4j.conf.ConfigurationBuilder;

@Controller
@RequestMapping("/tweets")
public class TweetsController {

	private volatile static List<Status> tweetList = new ArrayList<Status>();

	private static class _Run implements Runnable, Serializable {

		public void run() {
			Configuration conf = new ConfigurationBuilder()
					.setOAuthAccessToken(
							"2255068147-v32adeOq6b6wzM3KvnjusTHhuEykNxgkZ3PScyR")
					.setOAuthAccessTokenSecret(
							"vXkR9jtECj0CSPpH9lEyfsE8v8citcZ5N61VaqhKJll3n")
					.setOAuthConsumerKey("pcGABkK2IcOgImFtS20g")
					.setOAuthConsumerSecret(
							"v9KL2VMCvYslz5M7MLAu9DkVa3dcAEwNuSwOs5xU").build();
			Authorization auth = new OAuthAuthorization(conf);

			JavaStreamingContext ssc = new JavaStreamingContext(
					"spark://10.67.110.82:7077", "Tutorial2",
					new Duration(1000));
			JavaReceiverInputDStream<Status> tweets = TwitterUtils
					.createStream(ssc, auth);
			// ExtendedTwitterStream tweets = new
			// ExtendedTwitterStream(ssc.ssc());
			JavaDStream<String> statuses = tweets.filter(
					new Function<Status, Boolean>() {

						public Boolean call(Status v1) throws Exception {
							return v1.getGeoLocation() != null;
						}
					}).map(new Function<Status, String>() {

				public String call(Status v1) throws Exception {
					synchronized (tweetList) {
						if (tweetList.size() > 10000) {
							tweetList.remove(0);
						}
						tweetList.add(v1);
					}
					return v1.getText();
				}
			});
			tweets.print();
			ssc.start();
			ssc.awaitTermination();
		}
	}

	@PostConstruct
	public void init() {
		new Thread(new _Run()).start();
	}

	@RequestMapping("/get")
	@ResponseBody
	public List<Status> getStatus() {
		return tweetList;
	}

	public static void main(String[] args) {
		new _Run().run();
	}
}
